package com.swak.vertx.transport.vertx;

import com.swak.vertx.config.VertxConfig;
import com.swak.vertx.transport.VertxProxy;
import com.swak.vertx.transport.codec.Msg;
import com.swak.vertx.transport.codec.MsgCodec;
import com.swak.vertx.transport.vertx.btc.BlockedThreadEventHandler;

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.Message;
import io.vertx.core.impl.ContextInternal;
import io.vertx.core.impl.VertxImpl;

/**
 * Vertx 的配置 单机版本: 发起服务时必须处于 Vertx 的线程环境下，主要针对消息的回调，例如 MQ 的消费 <br>
 * context 用于非 Vertx 环境下的消息中转 <br>
 * 
 * @author: lifeng
 * @date: 2020/3/29 19:11
 */
public class StandardVertx implements VertxProxy {

	protected VertxConfig config;
	protected VertxImpl vertx;
	protected ContextInternal context;

	protected DeliveryOptions deliveryOptions;

	public StandardVertx(VertxConfig config) {
		this.config = config;
		this.deliveryOptions = new DeliveryOptions();
		this.deliveryOptions.setSendTimeout(config.getSendTimeout());
		this.deliveryOptions.setCodecName(MsgCodec.CODEC_NAME);
		this.vertx = (VertxImpl) Vertx.vertx(config);
		this.vertx.blockedThreadChecker().setThreadBlockedHandler(new BlockedThreadEventHandler());
		this.context = this.vertx.createEventLoopContext(this.vertx.getEventLoopGroup().next(), null,
				Thread.currentThread().getContextClassLoader());
	}

	/**
	 * 发送消息服务
	 */
	@Override
	public void sentMessage(String address, Msg request, int timeout) {
		this.sentMessage(address, request, timeout, null);
	}

	/**
	 * 发送消息服务
	 */
	@Override
	public void sentMessage(String address, Msg request, int timeout, Handler<AsyncResult<Message<Msg>>> replyHandler) {
		if (Vertx.currentContext() == null) {
			context.runOnContext((v) -> this.sentMessageInternal(address, request, timeout, replyHandler));
		} else {
			this.sentMessageInternal(address, request, timeout, replyHandler);
		}
	}

	/**
	 * 发送消息
	 */
	private void sentMessageInternal(String address, Msg request, int timeout,
			Handler<AsyncResult<Message<Msg>>> replyHandler) {
		DeliveryOptions deliveryOptions = this.deliveryOptions;
		if (timeout >= 1) {
			deliveryOptions = new DeliveryOptions(this.deliveryOptions);
			deliveryOptions.setSendTimeout(timeout);
		}
		vertx.eventBus().request(address, request, deliveryOptions, replyHandler);
	}

	@Override
	public Vertx me() {
		return this.vertx;
	}

	@Override
	public VertxConfig config() {
		return config;
	}
}